package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/25 18:03:49
 * Description: Todo
 * Version：1.0
 */
public class Dwd_03_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeCartAdd().init(30003,2,"Dwd_03_DwdTradeCartAdd");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // read ods_db data
        readOdsDb(tbe,"Dwd_03_DwdTradeCartAdd");
        // filter data that insert and update(sku_num was changed and new sku_num biggest than old) for cart_info
        Table result = tbe.sqlQuery(" select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['sku_id'] sku_id, " +
                " if(`type`='insert',cast(`data`['sku_num'] as int), " +
                " cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)) as sku_num, " +
                " ts " +
                " from ods_db " +
                " where `database`='gmall2023' and " +
                " `table`='cart_info' and " +
                " (`type`='insert' or " +
                " (`type`='update' and `data`['sku_num'] is not null and " +
                " cast(`data`['sku_num'] as int) > cast( `old`['sku_num'] as int) ) )");

        // put the result to kafka
        tbe.executeSql("create table dwd_trade_cart_add(" +
                " id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_num bigint, " +
                " ts bigint" +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
        result.executeInsert("dwd_trade_cart_add");
    }
}
